package climspace.hadoop;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;

@SuppressWarnings("deprecation")
public class GetHistoricalDataByDate {
	private String hostName;
	private String hdfsPort;
	private String mapredPort;
	
	public GetHistoricalDataByDate(String hostName, String hdfsPort, String mapredPort) {
		this.hostName = hostName;
		this.hdfsPort = hdfsPort;
		this.mapredPort = mapredPort;
	}

	public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		String fromDate = "";
		String toDate = "";
		
		public void configure(JobConf conf) {
			this.fromDate = conf.get("ClimSpace.Filter.FromDate");
			this.toDate = conf.get("ClimSpace.Filter.ToDate");
		}
		
		public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			// map will filter on date
			
			String line = value.toString();
			int intFromDate = Integer.parseInt(fromDate);
			int intToDate = Integer.parseInt(toDate);
			
			try {
				String measureDate = (line.substring(9, 17)).trim();
				int currentDate = Integer.parseInt(measureDate);

				if (intFromDate <= currentDate && intToDate >= currentDate) {
					output.collect(value, new Text());
				}
			} catch (NumberFormatException e) {
				System.out.println("Skipping non data lines");
			} catch (StringIndexOutOfBoundsException e) {
				System.out.println("Skipping non data lines");
			}
		}
	}

	public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {	
		public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output,
				Reporter reporter) throws IOException {
			output.collect(key, null);
		}
	}
	
	/**
	 * @param inputPath input folder on HDFS
	 * @param outputPath output folder on HDFS
	 * @param FromDate
	 * @param ToDate
	 * @throws Exception
	 */
	public void run(String inputPath, String outputPath, String FromDate, String ToDate) throws Exception {
		JobConf conf = new JobConf(GetHistoricalDataByDate.class);
		conf.setJobName("GetHistoricalDataByDate");

		conf.set("fs.default.name", "hdfs://" + hostName + ":" + hdfsPort + "/");
		conf.set("mapred.job.tracker", hostName + ":" + mapredPort);
		
		conf.set("ClimSpace.Filter.FromDate", FromDate);
		conf.set("ClimSpace.Filter.ToDate", ToDate);

		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		conf.setMapperClass(Map.class);
		// conf.setCombinerClass(Reduce.class);
		conf.setReducerClass(Reduce.class);

		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);

		Path output = new Path(outputPath);
		FileSystem fs = FileSystem.get(conf);
		fs.delete(output, true);
		
		FileInputFormat.setInputPaths(conf, new Path(inputPath));
		FileOutputFormat.setOutputPath(conf, output);
		
		//JobClient.runJob(conf);
		JobClient job = new JobClient(conf);
		job.submitJob(conf);
	}

	public static void main(String[] args) throws Exception {
		if (args.length != 4)
			throw new RuntimeException("Incorrent number of parameter");
		GetHistoricalDataByDate  wc = new GetHistoricalDataByDate("localhost", "9000", "9001");
		wc.run(args[0], args[1], args[2], args[3]);
	}

}
